package com.zuoye.weather

import org.apache.spark.SparkConf
import org.apache.spark.sql.{SaveMode, SparkSession}

object WeatherAnalysis {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "root")
    //-1. spark 配置对象
    val sparkConf = new SparkConf()
      .setMaster("local[*]")
    //-2. sparksession对象
    val sparkSession = SparkSession
      .builder()
      .config(sparkConf)
      .enableHiveSupport() //-:开启hive的支持
      .appName("WeatherAnalysis")
      .getOrCreate()
    // 读取hive表的数据
    try {
      // -3. 定义hdfs中文件的位置
      val originPath = "/weather-data/ods/sh.csv"
      // -4. 读取csv文件

      val df = sparkSession.read.option("header", "true").csv(originPath)
      // 读取CSV文件（假设文件名为weather.csv，且位于HDFS或本地文件系统中）
      // 注意：您需要根据实际情况修改文件路径和格式

      // 将DataFrame注册为临时视图以便进行SQL查询
      df.createTempView("weather_view")

      // 计算一些基本的统计信息

      // 1. 计算每种天气情况下的平均最高温度和最低温度
      val weatherTemps = sparkSession.sql(
        """
      SELECT weather, AVG(maxtemperature) AS avgMaxTemp, AVG(mintemperature) AS avgMinTemp
      FROM weather_view
      GROUP BY weather
    """)
      weatherTemps.write.mode(SaveMode.Overwrite).saveAsTable("db_weather.dws_weather_temps")

      // 2. 计算不同风向下的平均风力等级
      val windPowers = sparkSession.sql(
        """
      SELECT windDirection, AVG(windPower) AS avg_wind_power
      FROM weather_view
      GROUP BY windDirection
    """)
      windPowers.write.mode(SaveMode.Overwrite).saveAsTable("db_weather.dws_wind_powers")

      // 3. 计算不同空气质量指数标签下的天数
      val airQualityCounts = sparkSession.sql(
            """
         SELECT airQualityIndexLabel, COUNT(*) AS days_count
       FROM weather_view
        GROUP BY airQualityIndexLabel
      """)
       airQualityCounts.write.mode(SaveMode.Overwrite).saveAsTable("db_weather.dws_air_quality_counts")
        // 4.计算每种天气状况出现的次数
      val weather_count = sparkSession.sql("SELECT weather, COUNT(*) as count FROM weather_view GROUP BY weather")
      // 展示结果
      weather_count.show()
      // 将结果存储到Hive表中（如果需要）
      weather_count.write.mode(SaveMode.Overwrite).saveAsTable("db_weather.dws_weather_count_table")
      // 5.计算平均最高气温和平均最低气温
      val avg_temp = sparkSession.sql("SELECT AVG(maxtemperature) as avg_max_temp, AVG(mintemperature) as avg_min_temp FROM weather_view")
      // 展示结果
      avg_temp.show()
      // 将结果存储到Hive表中（如果需要）
      avg_temp.write.mode(SaveMode.Overwrite).saveAsTable("db_weather.dws_avg_temp_table")
      // 6.分析风向与天气状况的关系
      val wind_weather_relation = sparkSession.sql("SELECT windDirection, weather, COUNT(*) as count FROM weather_view GROUP BY windDirection, weather")
      // 展示结果
      wind_weather_relation.show()
      // 将结果存储到Hive表中（如果需要）
      wind_weather_relation.write.mode(SaveMode.Overwrite).saveAsTable("db_weather.dws_wind_weather_relation_table")
      // 7. 计算不同风向下的平均空气质量指数
      val airQualityIndex = sparkSession.sql(
        """
      SELECT windDirection, AVG(airQualityIndex) AS avg_quality_index
      FROM weather_view
      GROUP BY windDirection
    """)
      airQualityIndex.write.mode(SaveMode.Overwrite).saveAsTable("db_weather.dws_air_quality_index")
      // 8. 计算不同天气下的平均空气质量指数
      val weatherQualityIndex = sparkSession.sql(
        """
      SELECT weather, AVG(airQualityIndex) AS avg_quality_index
      FROM weather_view
      GROUP BY weather
    """)
      weatherQualityIndex.show()
      weatherQualityIndex.write.mode(SaveMode.Overwrite).saveAsTable("db_weather.dws_weather_quality_index")
      // sparkSession的停止、关闭
      sparkSession.stop()
      sparkSession.close()
    }
  }
}


